package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.MapTypeInfo;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Map;

/**
 * @author : GW00280745
 * @date : 2022/8/15 9:39
 * @description : 寻找电流map间隔点---条件是 电流变化幅度大于0.1
 */
public class MapBatterInterval implements Serializable {
    private String batterChangePercent;
    public MapBatterInterval(String batterChangePercent){
        this.batterChangePercent = batterChangePercent;
    }
    private MapStateDescriptor<String, Map<String, Object>> confInfo = new MapStateDescriptor<>("conf-info", BasicTypeInfo.STRING_TYPE_INFO, new MapTypeInfo<>(String.class, Object.class));

    public void process(JSONObject record,
                        MapState<String, String> latestBatterIntervalPointState,
                        KeyedBroadcastProcessFunction.ReadOnlyContext ctx){
        try {
            intervalPoint(record,latestBatterIntervalPointState,ctx);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    private void intervalPoint(JSONObject line,
                               MapState<String, String> latestBatterIntervalPointState,
                               KeyedBroadcastProcessFunction.ReadOnlyContext ctx) throws Exception{

        String frVal = line.getString(ChargePredictFields.ins().stdFieldChargeInitPoint);
        //首条记录是空值情况下
        if (StringUtils.isBlank(frVal)){
            line.put(ChargePredictFields.ins().stdFieldBatterIntervalPoint,ChargePredictFields.ins().zeroStr);
            return;
        }
        String chargeGroupName = line.getString(ChargePredictFields.ins().dwsFieldChargeBegin);
        Long tidVal = line.getLong(ChargePredictFields.ins().srcFieldTid);
        BigDecimal batter = line.getBigDecimal(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
        String s = tidVal.toString().concat(ChargePredictFields.ins().char3).concat(batter.toString());
        //有首条记录，那么判断状态中是否有数据，如果没有那么首条记录就是第一个间隔点
        if (!latestBatterIntervalPointState.contains(chargeGroupName)){
            latestBatterIntervalPointState.put(chargeGroupName,s);
            line.put(ChargePredictFields.ins().intervalPointFieldBatter,chargeGroupName);
            line.put(ChargePredictFields.ins().stdFieldBatterIntervalPoint,ChargePredictFields.ins().oneStr);
            return;
        }
        //状态中有数据，那么进行电流比较
        String latestTidBatter = latestBatterIntervalPointState.get(chargeGroupName);
        String latestBatterStr = StringUtils.substringAfterLast(latestTidBatter, ChargePredictFields.ins().char3);
        BigDecimal latestBatter = new BigDecimal(latestBatterStr);
        BigDecimal deltaBatter = batter.subtract(latestBatter).abs();
        boolean dcIsOne = StringUtils.equalsIgnoreCase(line.getString(ChargePredictFields.ins().srcFieldBmsDcConnect), "1");
        boolean obcIsOne = StringUtils.equalsIgnoreCase(line.getString(ChargePredictFields.ins().srcFieldBmsObcConnect), "1");
//        String electricQuantity = parseConf(ctx, BmsParamField.param_batterMap_intervalBatter, batterChangePercent)
        //电流map表 默认 = 1%
        String electricQuantity = "1";
        float v = batter.abs().floatValue();
        if (v>=num100){
            //快充 大于100那么采用0.1%。场景是=快充电流是梯度下降，每个梯度的电流变化特别小
            electricQuantity = "1";
        }else if (dcIsOne && v< num100 ){
            //小于100采用 1%
            electricQuantity = "1";
        }
        electricQuantity = "0.5";
        BigDecimal compareBaseVal = latestBatter.multiply(new BigDecimal(electricQuantity)).divide(new BigDecimal(100),BigDecimal.ROUND_HALF_DOWN).abs();
        //如果小于0说明当前变化幅度没在预定范围内
        if (deltaBatter.compareTo(compareBaseVal) < 0){
            line.put(ChargePredictFields.ins().intervalPointFieldBatter, ChargePredictFields.ins().blankStr);
            line.put(ChargePredictFields.ins().stdFieldBatterIntervalPoint,ChargePredictFields.ins().zeroStr);
            return;
        }
        //当前变化幅度在预定范围内 符合间隔点条件
        latestBatterIntervalPointState.put(chargeGroupName,s);
        line.put(ChargePredictFields.ins().intervalPointFieldBatter, deltaBatter.toString());
        line.put(ChargePredictFields.ins().stdFieldBatterIntervalPoint,ChargePredictFields.ins().oneStr);
    }
    private int num100 =100;
    private String parseConf(KeyedBroadcastProcessFunction.ReadOnlyContext ctx, String key, String defaultVal){
        try {
            ReadOnlyBroadcastState<String, Map<String, Object>> broadcastState = ctx.getBroadcastState(confInfo);
            Map<String, Object> confMap = broadcastState.get(null);
            return parseConfFromBroadcast(confMap,key,defaultVal);
        }catch (Exception e){
            e.printStackTrace();
            return defaultVal;
        }
    }
    private String parseConfFromBroadcast(Map<String, Object> confMap,String key,String defaultVal){
        try {
            Object confMapOrDefault = confMap.getOrDefault(key, defaultVal);
            return confMapOrDefault.toString();
        }catch (Exception e){
            return defaultVal;
        }
    }
}
